package com.ksyun.campus.metaserver.service.impl;

import com.alibaba.fastjson2.JSON;
import com.ksyun.campus.metaserver.common.constant.MqConstant;
import com.ksyun.campus.metaserver.common.result.BaseResult;
import com.ksyun.campus.metaserver.config.MasterManager;
import com.ksyun.campus.metaserver.entity.dto.req.WriteReqDto;
import com.ksyun.campus.metaserver.entity.meta.FileType;
import com.ksyun.campus.metaserver.entity.meta.ReplicaData;
import com.ksyun.campus.metaserver.entity.meta.StatInfo;
import com.ksyun.campus.metaserver.entity.meta.tree.FileTreeManager;
import com.ksyun.campus.metaserver.entity.server.DataServer;
import com.ksyun.campus.metaserver.mq.event.SyncMetaDataEvent;
import com.ksyun.campus.metaserver.mq.producer.SyncMetaDataProducer;
import com.ksyun.campus.metaserver.service.MetaService;
import com.ksyun.campus.metaserver.util.ZooKeeperUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;

/**
 * 元数据服务业务实现层
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class MetaServiceImpl implements MetaService {

    private final ZooKeeperUtil zooKeeperUtil;
    private final MasterManager masterManager;
    private final SyncMetaDataProducer syncMetaDataProducer;
    private final FileTreeManager fileTreeManager;
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public BaseResult<StatInfo> createFile(String fileSystemName, String path) {
        // 校验文件是否已经存在
        if (fileTreeManager.pathExists(fileSystemName, path)) {
            throw new RuntimeException(String.format("文件已存在，请勿重新创建 [path]: %s", path));
        }
        // 始化化文件副本信息
        List<ReplicaData> replicaDataList = new ArrayList<>();
        // 若无节点副本信息，则新增
        // 选择三个ds节点
        List<DataServer> allDataServer = getAllDataServer();
        if (allDataServer==null || allDataServer.isEmpty()) {
            throw new RuntimeException("无可用dataServer节点");
        }
        List<DataServer> threeDataServer;
        if (allDataServer.size() < 3) {
            threeDataServer = allDataServer;
        } else threeDataServer = allDataServer.subList(0, 3);

        int dataServerCount = threeDataServer.size();
        if (dataServerCount < 3) {
            log.info("dataServer可用副本节点数小于3");
        }
        for (int id = 1; id <= dataServerCount; id++) {
            DataServer dataServer = threeDataServer.get(id - 1);
            String ip = dataServer.getIp();
            Integer port = dataServer.getPort();
            // 构建副本信息
            replicaDataList.add(ReplicaData.builder()
                    .id(Integer.toString(id))
                    .dsNode(ip + ":" + port)
                    .path(path)
                    .build()
            );
        }
        // 创建文件元数据，并加入目录树（还没有文件大小）
        StatInfo statInfo = StatInfo.builder()
                .path(path)
                .type(FileType.File)
                .mtime(LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli())
                .replicaData(replicaDataList)
                .build();

        boolean createFileResult = fileTreeManager.addFile(fileSystemName, path, statInfo);
        if (!createFileResult) {
            throw new RuntimeException(String.format("父级目录不存在，拒绝创建 [path]: %s", path));
        }
        fileTreeManager.printFileTree(fileSystemName);
        // 发送同步请求
//        if (masterManager.getIsMaster()) {
        SyncMetaDataEvent syncMetaDataEvent = SyncMetaDataEvent.builder()
                .requestOption(MqConstant.CREATE)
                .requestPath("/meta/create")
                .fileSystemName(fileSystemName)
                .statInfo(statInfo)
                .path(path)
                .build();
        syncMetaDataProducer.send(syncMetaDataEvent);
//        }
        return BaseResult.success(statInfo);
    }

    @Override
    public BaseResult<String> commitWrite(String fileSystemName, WriteReqDto writeReqDto) {
        StatInfo statInfo = writeReqDto.getStatInfo();
        long size = statInfo.getSize();
        if (size == -1) {
            // 文件写入失败了 删除目录
            fileTreeManager.deleteFile(fileSystemName, statInfo.getPath());
            return BaseResult.success(String.format("文件写入失败，已删除文件 [path]:%s", statInfo.getPath()));
        }
        // 更新修改时间
        statInfo.setMtime(LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli());
        String path = statInfo.getPath();
        if (!fileTreeManager.updateFile(fileSystemName, path, statInfo)) {
            throw new RuntimeException(String.format("写文件提交失败 [path]: %s", path));
        }
        // 发送同步请求
//        if (masterManager.getIsMaster()) {
        SyncMetaDataEvent syncMetaDataEvent = SyncMetaDataEvent.builder()
                .requestOption(MqConstant.WRITE)
                .requestPath("/meta/write")
                .fileSystemName(fileSystemName)
                .statInfo(statInfo)
                .path(path)
                .build();
        syncMetaDataProducer.send(syncMetaDataEvent);
//        }
        fileTreeManager.printFileTree(fileSystemName);
        return BaseResult.success("写文件提交成功");
    }

    @Override
    public BaseResult<StatInfo> open(String fileSystemName, String path) {
        // 校验文件是否存在
        if (!fileTreeManager.pathExists(fileSystemName, path)) {
            throw new RuntimeException(String.format("文件不存在，请检查路径是否正确 [path]: %s", path));
        }
        StatInfo statInfo = fileTreeManager.findNode(fileSystemName, path);
        return BaseResult.success(statInfo);
    }

    @Override
    public BaseResult<StatInfo> mkdir(String fileSystemName, String path) {
        // 校验文件夹是否已经存在
        if (fileTreeManager.pathExists(fileSystemName, path)) {
            throw new RuntimeException(String.format("文件夹已存在，请勿重新创建 [path]: %s", path));
        }

        // 始化化文件副本信息
        List<ReplicaData> replicaDataList = new ArrayList<>();
        // 获取所有节点
        List<DataServer> allDataServer = getAllDataServer();
        int dataServerCount = allDataServer.size();
        if (dataServerCount < 3) {
            log.info("dataServer可用副本节点数小于3");
        }
        for (int id = 1; id <= dataServerCount; id++) {
            DataServer dataServer = allDataServer.get(id - 1);
            String ip = dataServer.getIp();
            Integer port = dataServer.getPort();
            // 构建副本信息
            replicaDataList.add(ReplicaData.builder()
                    .id(Integer.toString(id))
                    .dsNode(ip + ":" + port)
                    .path(path)
                    .build()
            );
        }
        StatInfo statInfo = StatInfo.builder()
                .path(path)
                .type(FileType.Directory)
                .mtime(LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli())
                // 文件夹不需要副本信息和大小
                .replicaData(replicaDataList)
                .size(0L)
                .build();
        boolean mkdirResult = fileTreeManager.addFile(fileSystemName, path, statInfo);
        if (!mkdirResult) {
            throw new RuntimeException(String.format("父级目录不存在，拒绝创建 [path]: %s", path));
        }
        fileTreeManager.printFileTree(fileSystemName);
        // 发送同步请求
//        if (masterManager.getIsMaster()) {
        SyncMetaDataEvent syncMetaDataEvent = SyncMetaDataEvent.builder()
                .requestOption(MqConstant.MKDIR)
                .requestPath("/meta/mkdir")
                .fileSystemName(fileSystemName)
                .statInfo(statInfo)
                .path(path)
                .build();
        syncMetaDataProducer.send(syncMetaDataEvent);
//        }
        return BaseResult.success(statInfo);
    }

    @Override
    public BaseResult<List<String>> listdir(String fileSystemName, String path) {
        StatInfo curNode = fileTreeManager.findNode(fileSystemName, path);
        if (curNode == null) {
            throw new RuntimeException(String.format("路径不存在，请检查路径是否正确 [path]: %s", path));
        }

        List<String> res = new ArrayList<>();

        try (Cursor<Map.Entry<Object, Object>> cursor = stringRedisTemplate.opsForHash().scan(fileSystemName, ScanOptions.scanOptions().match(path + "*").build())) {
            while (cursor.hasNext()) {
                Map.Entry<Object, Object> entry = cursor.next();
                String fullPath = (String) entry.getKey();

                // 排除自己
                if (path.equals(fullPath)) continue;

                // 计算相对路径
                String relativePath = fullPath.substring(path.length());
                if (relativePath.startsWith("/")) {
                    relativePath = relativePath.substring(1);
                }

                // 仅处理直接子目录
                if (relativePath.contains("/")) continue;

                String value = (String) entry.getValue();
                StatInfo statInfo = JSON.parseObject(value, StatInfo.class);
                res.add(statInfo.getPath());
            }
        } catch (Exception e) {
            throw new RuntimeException("获取文件夹异常", e);
        }

        return BaseResult.success(res);
    }

    @Override
    public BaseResult<StatInfo> delete(String fileSystemName, String path) {
        // 找到当前路径对应的节点
        // 暂存statInfo
        StatInfo statInfo = fileTreeManager.findNode(fileSystemName, path);
        if (statInfo == null) {
            throw new RuntimeException(String.format("路径不存在，请检查路径是否正确 [path]: %s", path));
        }
        if (!fileTreeManager.deleteFile(fileSystemName, path)) {
            throw new RuntimeException(String.format("文件或文件夹删除异常 [path]: %s", path));
        }

        // 发送同步请求
//        if (masterManager.getIsMaster()) {
        SyncMetaDataEvent syncMetaDataEvent = SyncMetaDataEvent.builder()
                .requestOption(MqConstant.DELETE)
                .requestPath("/meta/delete")
                .fileSystemName(fileSystemName)
                .statInfo(statInfo)
                .path(path)
                .build();
        syncMetaDataProducer.send(syncMetaDataEvent);
//        }
        fileTreeManager.printFileTree(fileSystemName);
        return BaseResult.success(statInfo);
    }

    @Override
    public BaseResult<StatInfo> getStats(String fileSystemName, String path) {
        // 找到当前路径对应的节点
        StatInfo statInfo = fileTreeManager.findNode(fileSystemName, path);
        if (statInfo == null) {
            throw new RuntimeException(String.format("文件不存在，请检查路径是否正确 [path]: %s", path));
        }
        return BaseResult.success(statInfo);
    }

    @Override
    public BaseResult<List<StatInfo>> listFileStats(String fileSystemName, String path) {
        StatInfo curNode = fileTreeManager.findNode(fileSystemName, path);
        if (curNode == null) {
            throw new RuntimeException(String.format("路径不存在，请检查路径是否正确 [path]: %s", path));
        }

        List<StatInfo> res = new ArrayList<>();

        try (Cursor<Map.Entry<Object, Object>> cursor = stringRedisTemplate.opsForHash().scan(fileSystemName, ScanOptions.scanOptions().match(path + "*").build())) {
            while (cursor.hasNext()) {
                Map.Entry<Object, Object> entry = cursor.next();
                String fullPath = (String) entry.getKey();

                // 排除自己
                if (path.equals(fullPath)) continue;

                // 计算相对路径
                String relativePath = fullPath.substring(path.length());
                if (relativePath.startsWith("/")) {
                    relativePath = relativePath.substring(1);
                }

                // 仅处理直接子目录
                if (relativePath.contains("/")) continue;

                String value = (String) entry.getValue();
                StatInfo statInfo = JSON.parseObject(value, StatInfo.class);
                if (statInfo.getType().equals(FileType.File)) {
                    res.add(statInfo);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("获取文件夹异常", e);
        }

        return BaseResult.success(res);
    }

    @Override
    public BaseResult<String> deleteErrorRecovery(String fileSystemName, StatInfo statInfo) {
        // 把元数据信息重新加入文件树
        fileTreeManager.addFile(fileSystemName, statInfo.getPath(), statInfo);
        // 发送同步请求
//        if (masterManager.getIsMaster()) {
        SyncMetaDataEvent syncMetaDataEvent = SyncMetaDataEvent.builder()
                .requestOption(MqConstant.DELETE_ERROR)
                .requestPath("/meta/create")
                .fileSystemName(fileSystemName)
                .statInfo(statInfo)
                .path(statInfo.getPath())
                .build();
        syncMetaDataProducer.send(syncMetaDataEvent);
//        }
        return BaseResult.success("删除失败回调恢复文件树成功");
    }

    /**
     * 通过zk内注册的ds列表，选择三个容量最大的dataServer
     *
     * @return ds节点的信息
     */
    public List<DataServer> getAllDataServer() {
        List<DataServer> dataServerMaplist = zooKeeperUtil.getDataServerlist(zooKeeperUtil.ZK_DATA_SERVER_ROOT_PATH);
        // 从注册中心拉取所有ds节点信息 按照剩余容量排序
        dataServerMaplist.sort(Comparator.comparingLong(ds -> ds.getUsed() - ds.getCapacity()));
        return dataServerMaplist;
    }
}
